package com.ztorn.cdc;

import com.ztorn.cdc.mysql.MysqlCDCBuilder;
import com.ztorn.common.assertion.Asserts;
import com.ztorn.common.model.FlinkCDCConfig;

import java.util.HashMap;
import java.util.Map;
import java.util.function.Supplier;

public class CDCBuilderFactory {

    private static final Map<String, Supplier<CDCBuilder>> CDC_BUILDER_MAP =
            new HashMap<String, Supplier<CDCBuilder>>() {
                {
                    put(MysqlCDCBuilder.METADATA_TYPE, () -> new MysqlCDCBuilder());
                }
            };

    public static CDCBuilder buildCDCBuilder(FlinkCDCConfig config) {
        if (Asserts.isNull(config) || Asserts.isNullString(config.getType())) {
            throw new RuntimeException("请指定 CDC Source 类型。");
        }
        return CDC_BUILDER_MAP
                .getOrDefault(
                        config.getType(),
                        () -> {
                            throw new RuntimeException(
                                    "未匹配到对应 CDC Source 类型的【" + config.getType() + "】。");
                        })
                .get()
                .create(config);
    }
}
